
/*
 * mpaxos.c
 *
 *  Created on: Dec 30, 2012
 *      Author: ms
 */

#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <json/json.h>
#include <apr_hash.h>
#include "mpaxos/mpaxos.h"
#include "proposer.h"
#include "acceptor.h"
#include "view.h"
#include "slot_mgr.h"
#include "utils/logger.h"
#include "utils/mtime.h"
#include "comm.h"

apr_pool_t *pool_ptr;
apr_hash_t *lastslot_ht_;   //groupid_t -> slotid_t
apr_hash_t *val_ht_;        //intid_t -> value_t
apr_hash_t *cb_ht_;         //groupid_t -> mpaxos_cb

pthread_mutex_t value_mutex;

int port_;

void mpaxos_init() {
    apr_initialize();
    apr_pool_create(&pool_ptr, NULL);
    lastslot_ht_ = apr_hash_make(pool_ptr);
    val_ht_ = apr_hash_make(pool_ptr);
    cb_ht_ = apr_hash_make(pool_ptr);

    // initialize view
    view_init();

    // initialize comm
    comm_init();

    // initialize proposer
    proposer_init();

    // initialize acceptor
    acceptor_init();
    
    // initialize slot manager
    slot_mgr_init();

    //start_server(port);

    pthread_mutex_init(&value_mutex, NULL);
}

void mpaxos_start() {
    start_server(port_);
}

void mpaxos_stop() {

}

void mpaxos_destroy() {
    acceptor_final();
    proposer_final();
    comm_final();


    //TODO stop eventloop
    stop_server();

    apr_pool_destroy(pool_ptr);
    apr_terminate();

    pthread_mutex_destroy(&value_mutex);
}

void mpaxos_set_cb(groupid_t gid, mpaxos_cb_t cb) {
    groupid_t *k = apr_palloc(pool_ptr, sizeof(gid));
    mpaxos_cb_t *v = apr_palloc(pool_ptr, sizeof(cb));
    *k = gid;
    *v = cb;
    apr_hash_set(cb_ht_, k, sizeof(gid), v);
}

void call_back(groupid_t gid, slotid_t sid, uint8_t* val, 
        size_t val_sz, void* para) {
    mpaxos_cb_t *cb = apr_hash_get(cb_ht_, &gid, sizeof(gid)); 
    (*cb)(gid, sid, val, val_sz, para);
}

void set_listen_port(int port) {
    port_ = port;
}

int commit_sync(groupid_t* gids, size_t gid_len, uint8_t *val,
        uint32_t val_len, void* cb_para) {

    // find the appropriate proposer
    // keep propose until success
    LOG_DEBUG("Try to commit.");
    struct timespec ts_begin;
    struct timespec ts_end;

    groupid_t gid = gids[0];
    roundid_t **rids_ptr;
    rids_ptr = (roundid_t **) malloc(gid_len * sizeof(roundid_t *));
    for (int i = 0; i < gid_len; i++) {
        rids_ptr[i] = (roundid_t *)malloc(sizeof(roundid_t));
        mpaxos__roundid_t__init(rids_ptr[i]);
        rids_ptr[i]->gid = (gids[i]);

        //get_insnum(gids[i], &sid_ptr);
        //*sid_ptr += 1;
        rids_ptr[i]->bid = (1);
        rids_ptr[i]->sid = acquire_slot(gids[i], get_local_nid());
    }
    do {
        get_realtime(&ts_begin);

        int rids_len = gid_len;
        int ret = run_full_round(gid, 1, rids_ptr, rids_len, val, val_len, 1000000);
        if (ret == 0) {
            get_realtime(&ts_end);
            long millisec = (ts_end.tv_sec - ts_begin.tv_sec) * 1000 + (ts_end.tv_nsec - ts_begin.tv_nsec) / 1000 / 1000;
            if (millisec > 100) {
                LOG_WARN("Value committed. Cost time too long: %d", millisec);
            } else {
                LOG_DEBUG("Value committed. Cost time: %d", millisec);
            }
            
            // remember and call back.
            for (int i = 0; i < rids_len; i++) {
                roundid_t *rid_ptr = rids_ptr[i];
                put_instval(rid_ptr->gid, rid_ptr->sid, val, val_len, cb_para);
            }

            break;
        } else if (ret == -1) {
            printf("Error in phase 1?\n");
        } else if (ret == -2) {
            printf("Error in phase 2?\n");
        }
    } while (0);

    for (int i = 0; i < gid_len; i++) {
        free(rids_ptr[i]);
    }
    free(rids_ptr);
    return 0;
}

int commit_async(groupid_t* gids, size_t gid_len, uint8_t *val,
        size_t val_len, void *cb) {
    //TODO
    return 0;
}

int get_insnum(groupid_t gid, slotid_t** in) {
    slotid_t* sid_ptr = apr_hash_get(lastslot_ht_, &gid, sizeof(gid));
    if (sid_ptr == NULL) {
        sid_ptr = apr_palloc(pool_ptr, sizeof(slotid_t));
        *sid_ptr = 0;
        groupid_t *gid_ptr = apr_palloc(pool_ptr, sizeof(groupid_t));
        *gid_ptr = gid;
        apr_hash_set(lastslot_ht_, gid_ptr, sizeof(gid), sid_ptr);
    }
    *in = sid_ptr;
    return 0;
}

int get_instval(uint32_t gid, uint32_t in, char* buf,
        uint32_t bufsize, uint32_t *val_size) {
    //TODO
    return 0;
}

bool has_value(groupid_t gid, slotid_t sid) {
    //TODO  
    return false;
}

int put_instval(groupid_t gid, slotid_t sid, uint8_t *data,
        size_t len, void *cb_para) {
    pthread_mutex_lock(&value_mutex);
    
    instid_t *iid_ptr = (instid_t *) malloc(sizeof(instid_t));
    iid_ptr->gid = gid;
    iid_ptr->sid = sid;
    value_t *val = (value_t *) malloc(sizeof(value_t));
    val->data = (uint8_t *) malloc(len);
    val->len = len;
    memcpy (val->data, data, len);

    // TODO just forget now.
//  apr_hash_set(val_ht_, iid_ptr, sizeof(instid_t), val);

    slotid_t *lastsid;
    get_insnum(gid, &lastsid);

    //TODO looping call back.
    while (has_value(gid, *lastsid + 1)) {
        call_back(gid, *lastsid + 1, data, len, cb_para);
        //cb_(gid, *lastsid + 1, data, len);
        *lastsid += 1;
    }

    //TODO free some space in the map.
    // forget
    acceptor_forget();

    pthread_mutex_unlock(&value_mutex);
    return 0;
}
